package com.spark.sql

import org.apache.spark.SparkConf
import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
import org.apache.spark.sql.types.{DataType, LongType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Row, SparkSession}

/**
  *
  * @author Pop
  * @date 2022/7/20 22:48
  */
object Spark03_SparkSql_UDAF {
  def main(args: Array[String]): Unit = {

    // 创建SparkSql的运行环境
    val sparkSql: SparkConf = new SparkConf().setMaster("local[*]").setAppName("sparkSql")
    val session: SparkSession = SparkSession.builder().config(sparkSql).getOrCreate()
    // 执行逻辑操作
      val df: DataFrame = session.read.json("datas/user.json")
    df.createOrReplaceTempView("user")

    // 希望如果你希望往结果定义一个前缀，可以使用自定义函数
    session.udf.register("ageA",new MyAvgUDAF())

    session.sql("select ageA(age) from user").show()

//      df.show
    // DataFrame =》 Sql
//    DataFrame2Sql(df,session)

  session.close()
  }

  /**
    * 自定义聚合函数类，计算年龄的平均值
    *
    * 1.继承 UserDefinedAggregateFunction
    * 2.重写发那个发
    */
  class MyAvgUDAF extends UserDefinedAggregateFunction{
    /**
      * 输入的结构，以上面的，我们要求年龄的平均值
      * 那么年龄的字段我们要指明类型
      * @return
      */
    override def inputSchema: StructType = {
      StructType(
        Array(
          StructField("age",LongType)
        )
      )
    }

    /**
      * 缓冲区结构
      * @return
      */
    override def bufferSchema: StructType = {
      StructType(
        Array(
          StructField("total",LongType),
          StructField("count",LongType)
        )
      )
    }

    /**
      * 函数计算结果的数据类型，Out
      * @return
      */
    override def dataType: DataType = LongType

    /**
      * 函数的稳定性，表示传入相同的参数，无论执行多少次，还是
      * 相同的结果
      * @return
      */
    override def deterministic: Boolean = true

    /**
      * 缓冲区初始化
      * @param buffer
      */
    override def initialize(buffer: MutableAggregationBuffer): Unit = {
//      buffer(0) = 0L
//      buffer(1) = 0L
      // 这里对应了，缓冲区的位置，因为我们定了两个字段，这些地方是要存放这些数值的
      /*
      * StructField("total",LongType),
          StructField("count",LongType)
      * */
      buffer.update(0,0L)
      buffer.update(1,0L)
    }

    /**
      * 根据输入的值，更新缓冲区数据
      * @param buffer
      * @param input
      */
    override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
      /**
        * 这里解释一下，input是我们定义的输入的值，由于我们只定义了age
        * 也就是只有一个位置，就是0位置的输入
        *
        * 然后就是缓冲区，我们缓冲区的结构定义成了两个
        * 所以存在0，1两个位置，当有新参数输入进来，这两个缓冲区
        * 结构数据的变化，情况，在这个方法中定义
        * */
      buffer.update(0,buffer.getLong(0)+input.getLong(0))
      buffer.update(1,buffer.getLong(1)+1)
    }

    /**
      * 缓冲区数据合并
      * @param buffer1
      * @param buffer2
      */
    override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
      //不断的累加结果
      buffer1.update(0,buffer1.getLong(0)+buffer2.getLong(0))
      buffer1.update(1,buffer1.getLong(1)+buffer2.getLong(1))
    }
    // 计算平均值
    override def evaluate(buffer: Row): Any = {
      //由于我们定义的是计算平均值，函数会调用这个返回最终结果
      //使用缓冲区的内容，来得到最后结果
      buffer.getLong(0)/buffer.getLong(1)
    }
  }
}
